#include <Kharlia/Core/Common.h>
#include <Kharlia/Core/IO.h>
#include <Kharlia/Core/App.h>
#include <Kharlia/Core/Tasker.h>

namespace kharlia { namespace core {

// True when a pending dispatch call has been added
volatile bool GPendingDispatchCall = false;
// When true, the file IO worker thread calls AddPendingDispatchCall()
volatile bool GThreadedAddPendingDispatchCall = false;

// Statistic collection and reporting
uint32 GStatDispatchCount = 0;
double GStatDispatchTotalTime = 0;

// Dispatch caller instance access
boost::mutex GIOManagerInstanceMutex;
CIOManager* CIOManager::msInstance = nullptr;
PyObject* CIOManager::msPyInstance = nullptr;

#define LOCK_IOM() boost::mutex::scoped_lock lock(GIOManagerInstanceMutex)

// Used by Py_MakePendingCalls()
int CIOManager::DispatchCaller(void* /*unused*/) {
    LOCK_IOM();
    if (CIOManager::msInstance) {
        try {
            CIOManager::msInstance->Dispatch(false);
        }
        catch (error_already_set&) {
            GPendingDispatchCall = false;
            return -1;
        }
        catch (std::exception& e) {
            if (!PyErr_Occurred())
                PyErr_Format(PyExc_RuntimeError, "Exception in dispatch caller: %s", e.what());
            GPendingDispatchCall = false;
            return -1;
        }
        catch (...) {
            if (!PyErr_Occurred())
                PyErr_SetString(PyExc_RuntimeError, "Unknown exception type in dispatch caller");
            GPendingDispatchCall = false;
            return -1;
        }
    }
    GPendingDispatchCall = false;
    return 0;
}

// Called by PyCObject destructor to handle the wrapped shared_ptr
void WrappedSharedPtrDestructor(void* obj) {
    shared_ptr<FWaitInfo>* wi = reinterpret_cast<shared_ptr<FWaitInfo>*>(obj);
    delete wi;
}

tuple Stat_ReportDispatch() {
    double avgTime = GStatDispatchTotalTime / double(GStatDispatchCount);
    return make_tuple(GStatDispatchCount, GStatDispatchTotalTime, avgTime);
}

void CIOManager::AddPendingDispatchCall(bool allowexc) {
    // Py_AddPendingCall is thread-safe, do not allow exceptions
    // if not calling from main thread
    if (GPendingDispatchCall)
        return;
    if (Py_AddPendingCall(&CIOManager::DispatchCaller, NULL) == -1) {
        if (allowexc) {
            throw RuntimeError("Failed to add pending call for IOManager Dispatcher");
        }
    }
    GPendingDispatchCall = true;
}

// This function is mostly for testing
// it will read on an already opened file in the future.
void FFileLoadRequest::Execute() {
    using namespace std;

    ifstream file;
    std::streamoff filesize;

    file.open(mFileName.c_str(), ios::binary | ios::ate);
    if (file.fail()) {
        mErrorCode = IOERR_OPENFAIL;
        throw exception();
    }
    // Set read length to length of file
    filesize = file.tellg();
    assert(filesize <= 0xFFFFFFFF);
    // TODO: 8 byte file size variable

    if (mReadLength < 1)
        mReadLength = (uint32)filesize;
    // Move back to the starting position
    file.seekg(mReadStart);

    // Reserve memory, read directly into the string, then trim any extra
    mResult.resize(mReadLength, '\0');
    file.read(const_cast<char*>(mResult.c_str()), mReadLength);
    assert(file.gcount() == mReadLength);
    mResult.resize((uint32)file.gcount());

    file.close();
}

void FFileDumpRequest::Execute() {
    using namespace std;

    ofstream file;
    file.open(mFileName.c_str(), ios::binary);
    if (file.fail()) {
        mErrorCode = IOERR_OPENFAIL;
        throw exception();
    }

    file.seekp(0);
    file.write(mDump.c_str(), mDump.size());
    file.close();
}

CIOManager::CIOManager(PyObject* self, uint8 poolsize):
    CSubsystem(self),
    mWorkerPool(poolsize),
    mDispatching(false)
{
}
CIOManager::~CIOManager() {
}

bool CIOManager::Init(object cls, uint8 poolsize) {
    LOCK_IOM();
    if (CIOManager::msInstance)
        return true;

    object inst = cls(poolsize);
    TPtr<CIOManager> cinst = extract<TPtr<CIOManager>>(inst);
    CIOManager::msInstance = cinst.GetPtr();
    CIOManager::msPyInstance = inst.ptr();
    Py_INCREF(CIOManager::msPyInstance);
    return true;
}

bool CIOManager::Exit() {
    LOCK_IOM();
    if (!CIOManager::msInstance)
        return true;

    CIOManager::msInstance = nullptr;
    Py_DECREF(CIOManager::msPyInstance);
    CIOManager::msPyInstance = nullptr;
    return true;
}

void CIOManager::Tick(float deltaTime) {
    Dispatch(true);
}

void CIOManager::Dispatch(bool async) {
    // Reentry guard added to prevent Dispatch() from being called again
    // by Python's pending calls while executing code to create a tasklet.
    if (mDispatching)
        return;
    mDispatching = true;
    try {
        double start = StatGetClock();
        // Only dispatch requests from this thread
        boost::thread::id thisThreadID = boost::this_thread::get_id();
        {
            // TODO: Add a counter for number of completed requests
            //       so we don't need to get a lock every time.
            boost::mutex::scoped_lock lock(mWaitingMutex);
            std::list<shared_ptr<FWaitInfo>>::iterator it;
            for (it = mWaitingQueue.begin(); it != mWaitingQueue.end(); it++) {
                // Only do a send if there is something receiving on the channel
                // in order to prevent a tasklet switch while the mutex is locked
                shared_ptr<FWaitInfo>& wir = *it;
                if ((wir->Request->GetState() == IORS_COMPLETED ||
                     wir->Request->GetState() == IORS_FAILED) && wir->ThreadID == thisThreadID) {
                    // Check that we have a stackless request and can send without blocking
                    if (wir->Channel && PyChannel_GetBalance(wir->Channel.get()) < 0) {
                        shared_ptr<FWaitInfo> wi = wir;
                        it = mWaitingQueue.erase(it);

                        int result = PyChannel_Send(wi->Channel.get(), Py_None);
                        wi->Request->mDispatched = true;

                        if (result == -1) {
                            if (!PyErr_Occurred())
                                PyErr_SetString(PyExc_RuntimeError,
                                    "Unknown error on channel send in CIOManager::Dispatch");
                            throw_error_already_set();
                        }

                        if (it == mWaitingQueue.end())
                            break;
                    }
                }
            }
        }
        double elapsed = StatGetClock() - start;
        GStatDispatchTotalTime += elapsed;
        GStatDispatchCount++;
    }
    catch (...) {
        mDispatching = false;
        throw;
    }
    mDispatching = false;
}

void CIOManager::Execute(intrusive_ptr<FIORequest> req) {
    try {
        req->mState = IORS_EXECUTING;
        req->Execute();
        req->mState = IORS_COMPLETED;
    }
    catch (std::exception&) {
        req->mState = IORS_FAILED;
        if (req->mErrorCode == IOERR_NONE)
            req->mErrorCode = IOERR_UNKNOWN;
    }
    if (!GPendingDispatchCall && GThreadedAddPendingDispatchCall && CIOManager::msInstance) {
        AddPendingDispatchCall();
    }
}

// Different types of IO requests will be handled differently.
// File IO requests are posted to a single worker thread (for now)
// that executes them in the order that they're posted.
void CIOManager::ExecuteSL(intrusive_ptr<FIORequest> req) {
    if (!req)
        throw ValueError("req is null");
    if (req->GetState() != IORS_READY)
        throw ValueError("request is not ready for execution");

    // FWaitInfo used by dispatcher to send on channels for
    // completed requests from the thread that made the request
    shared_ptr<FWaitInfo> wi = make_shared<FWaitInfo>();
    wi->Channel = handle<PyChannelObject>(PyChannel_New(NULL));
    // Set preference to 1 to prefer the sender, meaning the Dispatcher
    // can unblock many tasklets at once without a tasklet switch
    PyChannel_SetPreference(wi->Channel.get(), 1);
    wi->ThreadID = boost::this_thread::get_id();
    wi->Request = req;

    // Post FileIORequest to worker pool
    if (!mWorkerPool.schedule(boost::bind(&CIOManager::Execute, this, req))) {
        throw RuntimeError("unable to schedule IORequest");
    }

    // Now is a good time to dispatch anything that needs it
    Dispatch(false);

    // Check if the request completed while we were dispatching.
    if (!req->GetDispatched()) {
        {
            boost::mutex::scoped_lock lock(mWaitingMutex);
            mWaitingQueue.push_back(wi);
        }

        // Block until we're awakened by dispatch
        handle<> result(allow_null(PyChannel_Receive(wi->Channel.get())));

        // FWaitInfo removal is handled by the Dispatcher, so no need to worry
        // about it here.
        if (result.get() == NULL) {
            if (!PyErr_Occurred())
                PyErr_SetString(PyExc_RuntimeError,
                    "Unknown error on channel receive in CIOManager::ExecuteSL");
            throw_error_already_set();
        }
    }
}

// This is a temporary function for testing
str CIOManager::ReadFile(str filename, int start, int length) {
    std::string fn = extract<char const*>(filename);
    intrusive_ptr<FFileLoadRequest> req(new FFileLoadRequest(fn, start, length));
    ExecuteSL(req);
    return str(req->GetResult());
}

str CIOManager::LoadFile(str name) {
    intrusive_ptr<FFileLoadRequest> req(
        new FFileLoadRequest(extract<std::string>(name), 0, -1)
    );
    ExecuteSL(req);
    if (req->GetState() == IORS_FAILED)
        throw RuntimeError("load file failed");
    return str(req->GetResult());
}

void CIOManager::DumpFile(str name, str data) {
    intrusive_ptr<FFileDumpRequest> req(
        new FFileDumpRequest(
            extract<std::string>(name),
            extract<std::string>(data)
        )
    );
    ExecuteSL(req);
    if (req->GetState() == IORS_FAILED)
        throw RuntimeError("dump file failed");
}

KH_NO_GC(CIOManager)

void _InitIOBindings() {
    def("Stat_ReportDispatch", &Stat_ReportDispatch);
    class_<CIOManager, bases<CSubsystem>, noncopyable>
        ("CIOManagerBase", init<uint8>())
        .def("Dispatch", &CIOManager::Dispatch, (arg("async")=false))
        .def("ReadFile", &CIOManager::ReadFile)
        .def("LoadFile", &CIOManager::LoadFile)
        .def("DumpFile", &CIOManager::DumpFile)
        .def("Init", &CIOManager::Init, (arg("poolsize")=3)).staticmethod("Init")
        .def("Exit", &CIOManager::Exit).staticmethod("Exit")
        ;

    def("GetIOManager", &GetIOManager);

    KH_IMPL_GC(CIOManager);
    RegisterPtrConverter<CIOManager>();
}

}}